package com.dtease.agent.sync.extractor;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.baomidou.mybatisplus.extension.service.IService;
import com.dtease.agent.biz.agentstore.entity.BasToolsMetaDatabase;
import com.dtease.agent.biz.agentstore.service.IBasToolsMetaDatabaseService;
import com.dtease.agent.biz.resource.dto.BasStorageDTO;
import com.dtease.agent.biz.resource.entity.BasClusterService;
import com.dtease.agent.biz.resource.entity.RelClusterServiceDetail;
import com.dtease.agent.biz.resource.service.IBasClusterServiceService;
import com.dtease.agent.biz.resource.service.IRelClusterServiceDetailService;
import com.dtease.agent.common.utils.JdbcUtil;
import com.dtease.agent.common.utils.PasswordOperateUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import javax.sql.DataSource;
import java.util.List;

/**
 * @description Hive元数据提取器
 * @author chentl
 * @class HiveMetadataExtractor
 * @package com.dtease.agent.metadata.sync
 * @since 2025/5/7 14:23
 */
@Slf4j
@Component
public class HiveMetadataExtractor extends AbstractMetadataExtractor implements IMetadataExtractor {

    @Resource
    private IRelClusterServiceDetailService relClusterServiceDetailService;

    @Resource
    private IBasClusterServiceService basClusterServiceService;

    @Resource
    private IBasToolsMetaDatabaseService toolsMetaDatabaseService;

    @Override
    protected String getQuerySql(BasStorageDTO storage) {
        String dbName = getDatabaseName(storage);
        return "select t1.TBL_NAME as tableName, IFNULL(t5.param_value,'') as tableComment, " +
                " t4.column_name as columnName, t4.type_name as columnType, IFNULL(t4.comment,'') as columnComment " +
                " from tbls t1 left join dbs t2 on t2.DB_ID = t1.DB_ID " +
                " left join columns_v2 t4 on t4.CD_ID = t1.TBL_ID " +
                " left join table_params t5 on t5.tbl_id = t1.TBL_ID and t5.param_key = 'comment' " +
                " where t2.name = '" + dbName + "' AND (t5.PARAM_KEY not like '%flink.connector%' or t5.PARAM_KEY is null) " +
                " AND t4.column_name IS NOT NULL AND t4.type_name IS NOT NULL";
    }

    @Override
    protected DataSource getDataSource(BasStorageDTO storage) {
        JSONObject parameter = JSONUtil.parseObj(storage.getParameter());
        String clusterId = parameter.getStr("clusterId");

        List<BasClusterService> services = basClusterServiceService.selectByClusterId(clusterId);
        if (CollUtil.isEmpty(services)) {
            throw new RuntimeException("No Hive metastore service found for cluster ID: " + clusterId);
        }

        RelClusterServiceDetail detail = relClusterServiceDetailService.selectByClusterId(
                services.get(0).getClusterId(),
                services.get(0).getServiceId());

        JSONObject metaParams = JSONUtil.parseObj(detail.getParameter());
        String connectUrl = metaParams.getStr("javax.jdo.option.ConnectionURL");
        String username = metaParams.getStr("javax.jdo.option.ConnectionUserName");
        String password = PasswordOperateUtils.decrypt(metaParams.getStr("javax.jdo.option.ConnectionPassword"));
        String driver = metaParams.getStr("javax.jdo.option.ConnectionDriverName");

        return JdbcUtil.buildDataSource(connectUrl, username, password, driver);
    }

    @Override
    protected IService<BasToolsMetaDatabase> getToolsMetaDatabaseService() {
        return toolsMetaDatabaseService;
    }
}